package com.atbeijing.D09;

import com.atbeijing.D02.SensorReading;
import com.atbeijing.D02.SensorSource;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 * 使用状态变量实现分流温度平均值
 */
public class Example1 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env
                .addSource(new SensorSource())
                .keyBy(r -> r.id)
                .process(new KeyedProcessFunction<String, SensorReading, Tuple2<String,Double>>() {

                    //温度累加,个数
                    private transient ValueState<Tuple2<Double,Long>> state;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        super.open(parameters);
                        state=getRuntimeContext().getState(
                                new ValueStateDescriptor<Tuple2<Double,Long>>("state", Types.TUPLE(Types.STRING,Types.DOUBLE))
                        );
                    }

                    //每条数据处理一次
                    @Override
                    public void processElement(SensorReading value, Context ctx, Collector<Tuple2<String, Double>> out) throws Exception {
                        //第一次直接输出
                        if (state.value()==null){
                            state.update(Tuple2.of(value.temperature,1L));
                            out.collect(Tuple2.of(value.id,value.temperature));
                        }else {
                            //更新状态变量
                            state.update(Tuple2.of(value.temperature+state.value().f0,state.value().f1 +1L));
                            out.collect(Tuple2.of(value.id,state.value().f0/state.value().f1));
                        }
                    }
                })
                .print();


        env.execute();
    }
}
